/*
 * InterStage.java
 */

package pipeline;

import java.util.concurrent.Semaphore;


/**
 * A bounded buffer to be used as a pipeline interstage
 * 
 * @author Luca Molari
 *
 * @param <T> the interstage type.
 */
public class InterStage<T> implements I_DataSource<T>, I_DataSink<T> {

    protected final Object[] buffer;
	protected final int capacity;
	protected boolean closed;
    protected final Semaphore allowed_puts;
    protected int size;
    protected int next_write_index;
    protected int next_read_index;
	
	
	public InterStage(int capacity) {
		buffer = new Object[capacity];
		closed = false;
		this.capacity = capacity;
		allowed_puts = new Semaphore(capacity); 
		size = 0;
		next_write_index = 0;
		next_read_index = 0;
	}
	
	
	@Override
	@SuppressWarnings("unchecked")
	public T get() throws InterruptedException, SourceClosedException {
		T item;
		synchronized(this) {
			while (size == 0) {
				if (closed) {
					throw new SourceClosedException();
				}
				this.wait();
			}
			item = (T)buffer[next_read_index];
			next_read_index = (next_read_index + 1) % capacity;
			size -= 1;
		}
		allowed_puts.release();
		return item;
	}

	
	@Override
	public void close() {
		synchronized(this) {
			closed = true;
			/* wake up sleepers to get a SourceClosedException */
			this.notifyAll();
		}
	}

	
	@Override
	public void put(T d) throws InterruptedException {
		allowed_puts.acquire();
        synchronized (this) {
            buffer[next_write_index] = d;
            next_write_index = (next_write_index + 1) % capacity;
            ++size;
            /* wake up the getters */
            this.notifyAll();
        }
	}
	
	
}
